package com.shawn.integeration.kafka;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.stereotype.Component;

import java.util.Optional;

/**
 * @author shawn
 * @description: kafka 消息消费者
 * @date 2024/1/24
 */
@Slf4j
@Component
public class KafkaConsumer {
    // 消费主题
    private static final String TOPIC = "hello_kafka";
    // 消费组名
    private static final String TOPIC_GROUP = "test_consumer_group";

    @KafkaListener(topics = TOPIC, groupId = TOPIC_GROUP)
    public void topicTest(ConsumerRecord<?, ?> record, Acknowledgment ack,
                          @Header(KafkaHeaders.RECEIVED_TOPIC) String topic) {
        Optional<?> message = Optional.ofNullable(record.value());

        // 1. 判断消息是否存在
        if (!message.isPresent()) {
            return;
        }

        try {
            // 2. 消费消息
            Object msg = message.get();
            log.info("消费了 topic:[{}]消息:[{}]", topic, msg);

            // 3. 确认消费完成
            ack.acknowledge();
        } catch (Exception e) {
            log.error("消费消息失败 topic：{} message：{}", topic, message.get());
            throw e;
        }
    }
}
